package com.zp.qualitypractice.rx;

import android.database.Cursor;
import android.os.Bundle;
import android.support.design.widget.FloatingActionButton;
import android.support.design.widget.Snackbar;
import android.support.v7.app.AppCompatActivity;
import android.support.v7.widget.Toolbar;
import android.view.View;
import android.widget.Button;
import android.widget.EditText;

import com.zp.qualitypractice.ALog;
import com.zp.qualitypractice.App;
import com.zp.qualitypractice.Bean.Employee;
import com.zp.qualitypractice.Bean.ResponseBean;
import com.zp.qualitypractice.Bean.User;
import com.zp.qualitypractice.PhoneStateUtils;
import com.zp.qualitypractice.R;
import com.zp.qualitypractice.io.NetHelper;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import butterknife.BindView;
import butterknife.ButterKnife;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class RxBeginActivity extends AppCompatActivity {
    @BindView(R.id.toolbar)
    Toolbar toolbar;
    @BindView(R.id.etMain)
    EditText etMain;
    @BindView(R.id.button8)
    Button button8;
    @BindView(R.id.button9)
    Button button9;
    @BindView(R.id.button10)
    Button button10;
    @BindView(R.id.fab)
    FloatingActionButton fab;
    private String TAG = "RxBegin";
    private Subscription mSubscription;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_begin);
        ButterKnife.bind(this);
        fab.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                Snackbar.make(view, "Replace with your own action", Snackbar.LENGTH_LONG)
                        .setAction("Action", null).show();
            }
        });
//        allInMainThread();
//        observeTimes();
        button8.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                flowableAndSubscriber2();
            }
        });
        button9.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                request(1);
            }
        });

    }

    private void allInMainThread() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                ALog.v(TAG, "Observable thread is : " + Thread.currentThread().getName());
                ALog.v(TAG, "emit 1");
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                ALog.v(TAG, "Observer thread is :" + Thread.currentThread().getName());
                ALog.v(TAG, "onNext: " + integer);
            }
        };

        observable.subscribe(consumer);
    }

    private void observeOnMainThread() {

        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                ALog.v(TAG, "Observable thread is : " + Thread.currentThread().getName());
                ALog.v(TAG, "emit 1");
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                ALog.v(TAG, "Observer thread is :" + Thread.currentThread().getName());
                ALog.v(TAG, "onNext: " + integer);
            }
        };
//        先改变上游发送事件的线程, 让它去子线程中发送事件
        observable.subscribeOn(Schedulers.newThread())
//                然后再改变下游的线程, 让它去主线程接收事件
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);
    }

    private void observeTimes() {
        TAG = "observeTimes";
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                ALog.v(TAG, "Observable thread is : " + Thread.currentThread().getName());
                ALog.v(TAG, "emit 1");
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                ALog.v(TAG, "Observer thread is :" + Thread.currentThread().getName());
                ALog.v(TAG, "onNext: " + integer);
            }
        };
//        先改变上游发送事件的线程, 让它去子线程中发送事件
//        observable.subscribeOn(Schedulers.newThread())
//                .subscribeOn(Schedulers.io())
//                .observeOn(AndroidSchedulers.mainThread())
//                .observeOn(Schedulers.io())
//                .subscribe(consumer);
//        多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
//        多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.
        observable.subscribeOn(Schedulers.newThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        ALog.v(TAG, "After subscribeOn(newThread), current thread is: " + Thread.currentThread().getName());
                    }
                })
                .subscribeOn(Schedulers.io())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        ALog.v(TAG, "After subscribeOn(io), current thread is: " + Thread.currentThread().getName());
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        ALog.v(TAG, "After observeOn(mainThread), current thread is: " + Thread.currentThread().getName());
                    }
                })
                .observeOn(Schedulers.io())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        ALog.v(TAG, "After observeOn(io), current thread is : " + Thread.currentThread().getName());
                    }
                })
                .subscribe(consumer);
    }


    public Observable<List<Record>> readAllRecords() {
        return Observable.create(new ObservableOnSubscribe<List<Record>>() {
            @Override
            public void subscribe(ObservableEmitter<List<Record>> emitter) throws Exception {
                Cursor cursor = null;
                try {
//                    cursor = getReadableDatabase().rawQuery("select * from " + TABLE_NAME, new String[]{});
                    List<Record> result = new ArrayList<>();
//                    while (cursor.moveToNext()) {
//                        result.add(Db.Record.read(cursor));
//                    }
//                    在IO线程执行数据库操作，有了结果之后用onNext将ObservableOnSubscribe定义的泛型结果传递出去，至于谁接，不关心。
                    emitter.onNext(result);
//                    传递结果之后立即结束任务，这有点像new Thread执行一个任务。
                    emitter.onComplete();
                } finally {
                    if (cursor != null) {
                        cursor.close();
                    }
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }

    private class Record {

    }

    /**
     * 学习map方法对事件进行转换。
     */
    private void map() {
        TAG = "map";
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
//                连续发送三个整型的事件
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                //                利用map方法对整型事件进行接收和处理，所以Function要写明接收的类型和产出的类型
                ALog.d(TAG, integer);
                return "This is result " + integer;//这里产出String
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                ALog.d(TAG, s);
            }
        });
    }

    private void flatMap() {
        TAG = "flatMap";
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            //            flatMap和map的不同在于
//            map是转换数据为另外的数据。而flatMap是把可观察数据流类型转换为另外一种可观察数据流
//            map的function的产出参数是具体的数据类型，而flatMap的产出数据是ObservableSource，可观察的资源，当然也要定义观察的数据类型。
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
//                每接收到一个事件，就创建一个有三个元素的列表，把接收到的数据放到复制到三份转换为String数据，放到列表里。
                for (int i = 0; i < 3; i++) {
                    list.add("I the " + i + "value " + integer);
                }
//                然后遍历每次事件复制的三个String数据，通过from方法获取String并构造成ObservableSource<String>，延迟10毫秒发送出去

                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                ALog.d(TAG, s);
            }
        });
    }

    private void concatMap() {
        TAG = "concatMap";
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            //            concatMap吧, 它和flatMap的作用几乎一模一样, 只是它的结果是严格按照上游发送的顺序来发送的,
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I the " + i + "value " + integer);
                }
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                ALog.d(TAG, s);
            }
        });
    }

    private void combineRegisterALogin() {
        Map<String, String> params = new HashMap<>();
        params.put("mobile", "15736096668");
        params.put("captcha", "123456");
        params.put("password", "000000");
        params.put("source", "android");
        params.put("version", PhoneStateUtils.getAppVersionName(App.getAppContext()));
//        mMap.put("token", strToken);
//        mMap.put("mobile", strMobile);
        NetHelper.getInstance().getPaymentButler().register(params).subscribeOn(Schedulers.io())//在IO线程进行网络请求
                .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求注册结果
                .doOnNext(new Consumer<ResponseBean<List<User>>>() {
                    @Override
                    public void accept(@NonNull ResponseBean<List<User>> listResponseBean) throws Exception {
                        //先根据注册的响应结果去做一些操作
//                        当前应该是主线程，可以去调用一些主线程的方法
                        ALog.v(TAG, "current thread is: " + Thread.currentThread().getName());
//                        我在想，如果是这个时候出错的话，应该是调后面的哪个方法呢
//                        所以看来，这并不是一个好的方式。
                    }
                })
                .observeOn(Schedulers.io())                 //回到IO线程去发起登录请求
                .flatMap(new Function<ResponseBean<List<User>>, ObservableSource<ResponseBean<List<User>>>>() {
                    @Override
                    public ObservableSource<ResponseBean<List<User>>> apply(@NonNull ResponseBean<List<User>> listResponseBean) throws Exception {
                        Map<String, String> params = new HashMap<>();
                        params.put("mobile", "15736096668");
                        params.put("captcha", "123456");
                        params.put("password", "000000");
                        params.put("source", "android");
                        return NetHelper.getInstance().getPaymentButler().login(params);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求登录的结果
                .subscribe(new Consumer<ResponseBean<List<User>>>() {
                    @Override
                    public void accept(@NonNull ResponseBean<List<User>> listResponseBean) throws Exception {
                        ALog.d("登录成功");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        ALog.d("登录失败");
                    }
                });
    }

    private void learnZip() {
//        分别在两个io线程中创建事件发射器，连续发送事件
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                ALog.d(TAG, "emit 1");
                emitter.onNext(1);
                Thread.sleep(1000);

                ALog.d(TAG, "emit 2");
                emitter.onNext(2);
                Thread.sleep(1000);

                ALog.d(TAG, "emit 3");
                emitter.onNext(3);
                Thread.sleep(1000);

                ALog.d(TAG, "emit 4");
                emitter.onNext(4);
                Thread.sleep(1000);

                ALog.d(TAG, "emit complete1");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                ALog.d(TAG, "emit A");
                emitter.onNext("A");
                Thread.sleep(1000);

                ALog.d(TAG, "emit B");
                emitter.onNext("B");
                Thread.sleep(1000);

                ALog.d(TAG, "emit C");
                emitter.onNext("C");
                Thread.sleep(1000);

                ALog.d(TAG, "emit complete2");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());
//      用zip组合两个事件流，最后接到的事件数量，是上游流中事件数量最少的那个
//        不发送Complete呢？ 答案是显然的, 上游会继续发送事件, 但是下游仍然收不到那些多余的事件
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer + s;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                ALog.d(TAG, "onSubscribe");
            }

            @Override
            public void onNext(String value) {
                ALog.d(TAG, "onNext: " + value);
            }

            @Override
            public void onError(Throwable e) {
                ALog.d(TAG, "onError");
            }

            @Override
            public void onComplete() {
                ALog.d(TAG, "onComplete");
            }
        });
    }

    private void combineTwoSource() {
        Map<String, String> params = new HashMap<>();
        params.put("mobile", "15736096668");
        params.put("captcha", "123456");
        params.put("password", "000000");
        params.put("source", "android");
        params.put("version", PhoneStateUtils.getAppVersionName(App.getAppContext()));
        Observable<ResponseBean<List<User>>> observableUser = NetHelper.getInstance().getPaymentButler().login(params).subscribeOn(Schedulers.io());
        Observable<ResponseBean<List<Employee>>> observableEmployee = NetHelper.getInstance().getPaymentButler().getEmployees(params).subscribeOn(Schedulers.io());

//        还是那个问题，如果有异常怎么处理。
//        两个流都异常，两个流一个异常一个正常，等等情况怎么处理，是什么表现
        Observable.zip(observableUser, observableEmployee, new BiFunction<ResponseBean<List<User>>, ResponseBean<List<Employee>>, User>() {
            @Override
            public User apply(@NonNull ResponseBean<List<User>> listResponseBean,
                              @NonNull ResponseBean<List<Employee>> listResponseBean2) throws Exception {
                return null;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<User>() {
                    @Override
                    public void accept(@NonNull User user) throws Exception {

                    }
                });
    }

    /**
     * 连续多事件定量取一进行处理，使用filter过滤关键字
     */
    private void rationTakeOne() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io())
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
//                        如果发送的事件中integer能够被10整除，那么就filter让它过去。
                        return integer % 10 == 0;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        ALog.d(TAG, "" + integer);
                    }
                });
    }

    /**
     * 定时收取事件，不管事件多少。
     */
    private void timingTakeOne() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io())
//                sample操作符, 简单做个介绍, 这个操作符每隔指定的时间就从上游中取出一个事件发送给下游. 这里我们让它每隔2秒取一个事件给下游
                .sample(2, TimeUnit.SECONDS)  //sample取样
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        ALog.d(TAG, "" + integer);
                    }
                });
    }


    /**
     * 背压相关概念的基本理解。
     * 之前我们所的上游和下游分别是Observable和Observer, 这次不一样的是上游变成了Flowable, 下游变成了Subscriber,
     * 但是水管之间的连接还是通过subscribe()
     */
    private void flowableAndSubscriber() {
        Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
                                                         @Override
                                                         public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                                                             ALog.d(TAG, "emit 1");
                                                             emitter.onNext(1);
                                                             ALog.d(TAG, "emit 2");
                                                             emitter.onNext(2);
                                                             ALog.d(TAG, "emit 3");
                                                             emitter.onNext(3);
                                                             ALog.d(TAG, "emit complete");
                                                             emitter.onComplete();
                                                         }
                                                     },
//                首先是创建Flowable的时候增加了一个参数, 这个参数是用来选择背压,也就是出现上下游流速不均衡的时候应该怎么处理的办法,
// 这里我们直接用BackpressureStrategy.ERROR这种方式, 这种方式会在出现上下游流速不均衡的时候直接抛出一个异常,
// 这个异常就是著名的MissingBackpressureException.
                BackpressureStrategy.ERROR); //增加了一个参数
//分别创建了一个上游Flowable和下游Subscriber, 上下游工作在同一个线程中
        Subscriber<Integer> downstream = new Subscriber<Integer>() {

            @Override
            public void onSubscribe(Subscription s) {
//                下游的onSubscribe方法中传给我们的不再是Disposable了, 而是Subscription,
                ALog.d(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);  //注意这句代码
//                之前我们说调用Disposable.dispose()方法可以切断水管, 同样的调用Subscription.cancel()也可以切断水管,
// 不同的地方在于Subscription增加了一个void request(long n)方法

//                request当做是一种能力, 当成下游处理事件的能力, 下游能处理几个就告诉上游我要几个,
// 这样只要上游根据下游的处理能力来决定发送多少事件, 就不会造成一窝蜂的发出一堆事件来, 从而导致OOM.
            }

            @Override
            public void onNext(Integer integer) {
                ALog.d(TAG, "onNext: " + integer);

            }

            @Override
            public void onError(Throwable t) {
                ALog.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {
                ALog.d(TAG, "onComplete");
            }
        };
//        连接上下游
//        上下游工作在同一个线程, 上游每发送一个事件应该会等待下游处理完了才会继续发事件啊, 不可能出现上下游流速不均衡
        upstream
//                .subscribeOn(Schedulers.io())
//                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(downstream);
    }

    /**
     * @param n
     * 动态的向上游请求事件，一个或者多个。
     */
    public void request(long n) {
        mSubscription.request(n); //在外部调用request请求上游
//        如果这里不请求，那么下游就不会处理任何事件，事件全部阻塞在缓存中
//        异步请求就算下游每处理事情，也不回立即报错，存到129个事件才报错。
    }

    /**
     * 动态请求背压数据。
     */
    public void flowableAndSubscriber2() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                ALog.d(TAG, "emit 1");
                emitter.onNext(1);
                ALog.d(TAG, "emit 2");
                emitter.onNext(2);
                ALog.d(TAG, "emit 3");
                emitter.onNext(3);
                ALog.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {

                    @Override
                    public void onSubscribe(Subscription s) {
                        ALog.d(TAG, "onSubscribe");
                        mSubscription = s;  //把Subscription保存起来
                    }

                    @Override
                    public void onNext(Integer integer) {
                        ALog.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        ALog.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        ALog.d(TAG, "onComplete");
                    }
                });
    }

    private void backpressureStrategy(int strategy){
        switch (strategy){
            case 0:
                Flowable.create(new FlowableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                        for (int i = 0; ; i++) {
                            emitter.onNext(i);
                        }
                    }
                }, BackpressureStrategy.LATEST).subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(new Subscriber<Integer>() {

                            @Override
                            public void onSubscribe(Subscription s) {
                                ALog.d(TAG, "onSubscribe");
                                mSubscription = s;
                            }

                            @Override
                            public void onNext(Integer integer) {
                                ALog.d(TAG, "onNext: " + integer);
                            }

                            @Override
                            public void onError(Throwable t) {
                                ALog.w(TAG, "onError: ", t);
                            }

                            @Override
                            public void onComplete() {
                                ALog.d(TAG, "onComplete");
                            }
                        });
                break;
            case 1:
                Flowable.create(new FlowableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                        for (int i = 0; i < 10000; i++) {  //只发1w个事件
                            emitter.onNext(i);
                        }
                    }
                }, BackpressureStrategy.DROP).subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(new Subscriber<Integer>() {

                            @Override
                            public void onSubscribe(Subscription s) {
                                ALog.d(TAG, "onSubscribe");
                                mSubscription = s;
                                s.request(128);  //一开始就处理掉128个事件
                            }

                            @Override
                            public void onNext(Integer integer) {
                                ALog.d(TAG, "onNext: " + integer);
                            }

                            @Override
                            public void onError(Throwable t) {
                                ALog.w(TAG, "onError: ", t);
                            }

                            @Override
                            public void onComplete() {
                                ALog.d(TAG, "onComplete");
                            }
                        });
                break;
            case 2:
                Flowable.interval(1, TimeUnit.MICROSECONDS)
                        .onBackpressureDrop()  //加上背压策略
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(new Subscriber<Long>() {
                            @Override
                            public void onSubscribe(Subscription s) {
                                ALog.d(TAG, "onSubscribe");
                                mSubscription = s;
                                s.request(Long.MAX_VALUE);
                            }

                            @Override
                            public void onNext(Long aLong) {
                                ALog.d(TAG, "onNext: " + aLong);
                                try {
                                    Thread.sleep(1000);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }

                            @Override
                            public void onError(Throwable t) {
                                ALog.w(TAG, "onError: ", t);
                            }

                            @Override
                            public void onComplete() {
                                ALog.d(TAG, "onComplete");
                            }
                        });
                break;
        }

    }
}
